package com.merge.task;

import com.common.FileWriterUtils;
import com.common.MinHeap;
import com.merge.FileWrite;
import com.merge.constant.FileConstant;
import com.merge.enums.TaskStatus;
import com.merge.pojo.Data;
import com.merge.queue.SliceQueue;
import com.merge.queue.SliceQueueManger;

/**
 * 合并任务
 */
public class MergeTask extends Task {

    private String mergeFileName = FileConstant.MERGE_FILE_NAME;

    private MinHeap minHeap;

    public MergeTask(int id) {
        super(id, "merge_task");
    }

    @Override
    public void run() {
        //构建最小堆
        String mergeFilePath = getMergeFilePath();
        Data[] values = SliceQueueManger.allSliceQueuePollOne();
        minHeap = MinHeap.construct(values);
        while(getStatus() == TaskStatus.RUN.getValue()) {
            try {
                Thread.sleep(0);
                if (minHeap.empty()) {
                    FileWriterUtils.flush(mergeFilePath);
                    status = TaskStatus.STOP.getValue();
                    System.out.println("合并任务完成");
                    break;
                }
                Data value = minHeap.getMinValue();
                if (value == null) {
                    continue;
                }
                //System.out.println(value.getValue());
                FileWrite.writeLine(mergeFilePath, value.getValue());
                SliceQueue<Data> queue = SliceQueueManger.getQueueForName(value.getPipelineQueueNmae());
                Data data = queue.poll();
                if (data != null) {
                    minHeap.addValue(data);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public String getMergeFilePath() {
        return FileConstant.FILE_PATH + "\\" + mergeFileName + FileConstant.FILE_SUFFIX;
    }
}
